package org.niit.service

import org.apache.hadoop.hbase.client.Get
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.ml.recommendation.ALSModel
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Answer
import org.niit.util.{HBaseUtil, SparkUtil}

/*
推荐易错题服务
 */
class EDURecommendService {

  val spark = SparkUtil.takeSpark()
  import spark.implicits._   //导入Spark SQL中所需要的依赖
  import org.apache.spark.sql.functions._

  def dataAnalysis(answer:DStream[Answer]): Unit ={

    answer.foreachRDD(rdd=>{
      //1.获取模型
          //1.1如果模型路径存放再HBase中用下面的方法
      HBaseUtil.setHTable("bigdata:student")
      //获取模型路径
      val value:List[String] = HBaseUtil.getData( new Get( Bytes.toBytes("als_model-recommended_question_id") ) )
      val path = value(0)
          //1.2如果模型没有存入HBase，而是训练好的用下面的方法
      // val path = "E:\\大三下\\BD1\\BD1-Spark\\output\\als_model\\1681293896639"
      //2.加载模型
      val model = ALSModel.load(path)
      //由于训练模型的时候，不能有任何中文字样，所在在训练的时候，将id进行了切割：： 学生ID_45 == 45

      //将用户id转为数字的UDF(Spark SQL中自定义方法),因为后续模型推荐的时候需要用到数字格式id
      val id2Int = udf((student_id:String)=>{
        student_id.split("_")(1).toInt
      })
      //3.由于ALS模型传入的参数必须是DataFrame，所以要将RDD转成DataFrame
      val answerDF:DataFrame =  rdd.toDF
      //由于是针对学生Id进行推进错题，所以先将student查询并切割出来
      val studentIdDF =  answerDF.select(id2Int('student_id) as "student_id" )
      //4.使用模型给学生推荐题目    studentIdDF：要推荐的学生ID         10:推荐10道题目
      val recommendDF = model.recommendForUserSubset(studentIdDF, 10)
      //将推荐的数据结构进行打印
      recommendDF.printSchema()
      recommendDF.show(false) //false将省略的信息也一并打印出来  true:…………
      //5.处理推荐结果，取出用户id和题目id，将题目拼接成一个字符串 [id1,id2,....id10]
      val res = recommendDF.as[ (Int,Array[(Int,Float)]) ].map(t=>{
        //模型中储存都是数字，后面要存进数据库的时候，要和 学生ID_ 进行拼接
        val studentId: String = "学生ID_" + t._1
        val questionId = t._2.map("题目ID_" + _._1).mkString(",")
        (studentId,questionId)
      }).toDF("student_id","recommendations")
      //6.将answerDF 和  res 进行 join 合并
      val allInfoDF =  answerDF.join(res,"student_id")
      //7.保存到数据库当中
      allInfoDF.write
        .format("jdbc")   //连接MySQL用的是JDBC Oracle ODBC
        .option("url","jdbc:mysql://node1:3306/BD1?useUnicode=true&characterEncoding=utf8") //链接地址 BD1:数据库名 node1：数据库所在的IP地址 3306：端口号
        .option("driver","com.mysql.jdbc.Driver") //数据库驱动
        .option("user","root")   //数据库的用户名
        .option("password","Niit@123") //数据库的密码
        .option("dbtable","edu") //数据库表
        .mode(SaveMode.Append)//当该数据表不存在的时候，会自动的创建
        .save()

    })

  }

}
